package com.we.risk.phoneInfo.ato;

import com.alibaba.fastjson.JSONObject;
import com.we.doris.SinkDoris;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.utils.iplocation.IPLocation;
import com.we.utils.iplocation.Location;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.Map;
import java.util.Properties;

import static com.we.risk.phoneInfo.ato.commonVars.*;

public class phoneInfo {
    public static final String RELEASEPROP =
            "risk/phoneinfo/ato/kfk_prod.properties";
    public static Logger LOG = LoggerFactory.getLogger(phoneInfo.class);
    public static final String KFKBROKERHOSTBAK =
            "10.10.13.22:9092,10.10.13.23:9092,10.10.13.24:9092";
    public static final String KFKBAKTOPIC = "kafka.topic.data.record";
    public static final String KFKBAKTOPICGRPID = "Ato_phone_info_bak_grp";

    public static final Long KFKMSGMAXSIZE = 200 * 1024 * 1024L;

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(5 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            /** set offset to latest */
            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.latest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            consumProp.setProperty("group.id", KFKBAKTOPICGRPID);
            KafkaSource<String> kfkBakSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(KFKBROKERHOSTBAK)
                            .setTopics(KFKBAKTOPIC)
                            .setStartingOffsets(OffsetsInitializer.latest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            env.registerCachedFile(IPDATA, IPDATACATCHFILENAME);

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "PhoneRecord")
                            .uid("kfk-phoneinfo-source");
            SingleOutputStreamOperator<String> inputBak =
                    env.fromSource(
                                    kfkBakSource,
                                    WatermarkStrategy.noWatermarks(),
                                    "Phoneinfo-bak")
                            .uid("kfk-Phoneinfo-bak-source");

            DataStream<String> unionInput = input.union(inputBak);

            SingleOutputStreamOperator<String> out = unionInput.flatMap(new formatPhoneInfo())
                    .keyBy(new KeySelector<JSONObject, String>() {
                        @Override
                        public String getKey(JSONObject value) throws Exception {
                            return value.getString(OUTPUTKEY);
                        }
                    }).flatMap(new AtoPhoneInfo());

            DorisSink<String> dorisSink = SinkDoris.buidJsonDorisSink(FENODES, TABLEIDENTIFIER,
                    LABEL+System.currentTimeMillis(),
                    USERNAME, PASSWORD, SEQUENCECOL);
            out.sinkTo(dorisSink).setParallelism(1);

            env.execute(phoneInfo.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class formatPhoneInfo extends RichFlatMapFunction<String, JSONObject> {
        @Override
        public void flatMap(String input, Collector<JSONObject> out) throws Exception {
            try {
                if (input.contains(MSGTYPE)) {
                    JSONObject inputJson = JSONObject.parseObject(input);
                    if (inputJson != null && inputJson.getString("type").equals(MSGTYPE)) {
                        JSONObject dataJson = inputJson.getJSONObject("dataJson");
                        JSONObject job = dataJson.getJSONObject("job");
                        String jobID = job.getString(JOBID);

                        String userKey = dataJson.getString(INPUTKEYBY);
                        if (userKey != null) {
                            JSONObject outJson = new JSONObject();

                            JSONObject data = dataJson.getJSONObject("data");
                            JSONObject baseData = data.getJSONObject("baseData");
                            Long recv_time = null;
                            if(baseData != null) {
                                recv_time = baseData.getLong(HTTPTIMESTAMP);
                                outJson.put(OUTPUTKEY, userKey);
                                outJson.put(RECVTIME, recv_time);
                                outJson.put(OUTJOBID, jobID);
                                outJson.put(CELLIP, data.getString("cellIp"));
                                outJson.put(CELLIPCITY, data.getString("cellIPCity"));
                                outJson.put(CELLIPPROV, data.getString("cellIPProvince"));
                                outJson.put(DEVICEID, data.getString("deviceId"));
                                outJson.put(OS, data.getString("os"));
                                outJson.put(PALTFORM, data.getString("platform"));
                                outJson.put(JAILBREAK,(data.getBooleanValue("jailbreak") ? 1 : 0));
                                outJson.put(ROOT,(data.getBooleanValue("root") ? 1 : 0));
                                out.collect(outJson);
                            }
                        }
                    }
                }
            } catch (Exception e) {
                LOG.error("formatPhoneInfo excep: " + e.getMessage());
                LOG.error("formatPhoneInfo excep input: " + input);
                e.printStackTrace(System.out);
            }
        }
    }

    private static class AtoPhoneInfo extends RichFlatMapFunction<JSONObject, String> {
        ValueState<Long> recvTimeVState;
        IPLocation ipLocation;

        @Override
        public void open(Configuration parameters) throws Exception {
            recvTimeVState = getRuntimeContext()
                    .getState(new ValueStateDescriptor<Long>("RecvTimeVState", Long.class));
            initIpDataFile();
        }

        @Override
        public void flatMap(JSONObject value, Collector<String> out) throws Exception {
            try {
                LOG.debug("AtoPhoneInfo input: "+value.toString());
                Long tmpState = recvTimeVState.value();
                long curState = (tmpState == null ? 0L : tmpState);

                Long recv_time = value.getLong(RECVTIME);
                if (recv_time > curState) {
                    String cellIP = value.getString(CELLIP);
                    Map<String, String> ipMap = locateIpMap(cellIP);
                    LOG.debug("##ipmap:" + ipMap.toString());
                    if(ipMap != null) {
                        value.put(CITYID, ipMap.get(CITYID));
                        out.collect(value.toString());
                    }
                    recvTimeVState.update(recv_time);
                }
            } catch (Exception e) {
                LOG.error("AtoPhoneInfo excep: " + e.getMessage());
                LOG.error("AtoPhoneInfo excep input: " + value.toString());
            }
        }

        private void initIpDataFile() throws Exception{
            File ipDataFile = getRuntimeContext().getDistributedCache()
                            .getFile(IPDATACATCHFILENAME);
            FileInputStream fileInputStream = new FileInputStream(ipDataFile);
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fileInputStream));

            try{
                byte[] b = new byte[1024];
                ByteArrayOutputStream out = new ByteArrayOutputStream();
                while(fileInputStream.read(b) != -1){
                    out.write(b);
                }
                ipLocation = new IPLocation(out.toByteArray());
            }catch (Exception e) {
                LOG.error("initIpDataFile excep: " + e.getMessage());
            } finally {
                bufferedReader.close();
                fileInputStream.close();
            }
        }

        private Map<String, String> locateIpMap(String ip) throws Exception {
            if (ip == null || ip.equals("")) {
                return null;
            }
            Location location = ipLocation.fetchIPLocation(ip);
            return location.toMap();
        }
    }

}
